package cn.jltx.rxjava.rx.learn;

import android.util.Log;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

/**
 * @author jltxseo
 * Created by junlintianxia on 2017/08/09.
 */

public class RxJava2Learn2 {

    public static void justOperator() {
        Observable.just("map", "map1", "map2")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + s);
                    }
                });
    }

    public static void fromArrayOperator() {
        Observable.fromArray(new String[]{"map", "map1", "map2"})
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + s);
                    }
                });
    }

    /**
     * 接收一个 Iterable 容器作为输入，每次发射一个元素
     */
    public static void fromIterableOperator() {
        List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(1);
        list.add(20);

        Observable.fromIterable(list)
                .subscribe(num -> Log.d(XqTag.TAG, "accept:" + num));
    }

    /**
     * map的作用就变换 Observable 然后返回一个指定类型的 Observable 对象
     */
    public static void mapOperator() {
        Observable.just("map", "map1", "map2")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s + "-jltx";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + s);
                    }
                });
    }

    /**
     * map 操作符更神奇的地方是，你可以返回任意类型的 Observable，
     * 也就是说你可以使用 map 操作符发射一个新的数据类型的 Observable 对象。
     */
    public static void mapOperator1() {
        Observable.just("map", "map1", "map2")
                .map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String s) throws Exception {
                        return s.hashCode();
                    }
                })
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        return integer.toString() + "-jltx";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + s);
                    }
                });
    }

    /**
     * Flowable.flatMap 可以把一个 Observable 转换成多个 Observable
     */
    public static void flatMapOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        Observable.just(list1, list2, list3, list4)
                .flatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG, "中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        int delayTime = (int) (1 + Math.random() * 20);
                        return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integers) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + integers);
                    }
                });
    }


    public static void concatMapOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        Observable.just(list1, list2, list3, list4)
                .concatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG, "中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        int delayTime = (int) (1 + Math.random() * 20);
                        return Observable.fromIterable(integers).delay(delayTime, TimeUnit.MILLISECONDS);
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integers) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + integers);
                    }
                });
    }


    public static void flatMapIterableOperator() {
        List<Integer> list1 = new ArrayList<>();
        list1.add(1);
        list1.add(2);
        list1.add(3);

        List<Integer> list2 = new ArrayList<>();
        list2.add(4);
        list2.add(5);
        list2.add(6);

        List<Integer> list3 = new ArrayList<>();
        list3.add(7);
        list3.add(8);
        list3.add(9);

        List<Integer> list4 = new ArrayList<>();
        list4.add(10);
        list4.add(11);
        list4.add(12);

        List<List<Integer>> list = new ArrayList<>();
        list.add(list1);
        list.add(list2);
        list.add(list3);
        list.add(list4);

        Observable.fromIterable(list)
                .flatMapIterable(new Function<List<Integer>, Iterable<Integer>>() {
                    @Override
                    public Iterable<Integer> apply(List<Integer> integers) throws Exception {
                        Log.d(XqTag.TAG, "中间变换apply:" + integers.toString());
                        //使用一个随机的演示验证事件的顺序
                        return integers;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + integer);
                    }
                });
    }

    /**
     * filter 是用于过滤数据的，返回false表示拦截此数据。
     */
    public static void filterOperator() {
        Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer.intValue() > 5;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(XqTag.TAG, "accept:" + integer);
                    }
                });
    }

    /**
     * take 用于指定订阅者最多收到多少数据
     */
    public static void takeOperator() {
        Observable.fromArray(1, 10, 6, 0, 20, -1, 8)
                .take(10)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(Integer value) {
                        Log.d(XqTag.TAG, "onNext:" + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete");
                    }
                });
    }

    /**
     * 按照两个Observable提交结果的时间顺序，对Observable进行合并
     */
    public static void mergeOperator() {

        //被订阅者每隔1000毫秒后开始发射数据，并只取前面5个数据 发射顺序为0 2 4 6 8
        Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(5);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);
        Observable.merge(observable1, observable2).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(Long value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

    /**
     * 按照两个Observable提交结果的时间顺序，对Observable进行合并
     * 一旦某一个的Observable中出现错误就会马上停止合并，并对下游的订阅者执行onError方法。
     */
    public static void mergeOperatorErro() {

        //抛错误的Observable
        Observable errorObservable = Observable.error(new Exception("this is end!")).delay(3500, TimeUnit.MILLISECONDS);

        //被订阅者每隔1000毫秒后开始发射数据，并只取前面5个数据 发射顺序为0 2 4 6 8 第3500毫秒后产生错误
        Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(5).mergeWith(errorObservable);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);
        Observable.merge(observable1, observable2).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(Long value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }


    /**
     * mergeDelayError演示代码
     */
    public static void mergeDelayErrorOperatorErro() {

        //抛错误的Observable
        Observable errorObservable = Observable.error(new Exception("this is end!")).delay(3500, TimeUnit.MILLISECONDS);

        //被订阅者每隔1000毫秒后开始发射数据，并只取前面5个数据 发射顺序为0 2 4 6 8 第3500毫秒后产生错误
        Observable observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(5).mergeWith(errorObservable);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);
        Observable.mergeDelayError(observable1, observable2).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(Long value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

    /**
     * zip操作符演示
     */
    public static void zipOperator() {

        //被订阅者每隔1000毫秒后开始发射数据，并只取前面3个数据 发射顺序为0 2 4
        Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(3);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);

        Observable.zip(observable1, observable2, new BiFunction<Long, Long, String>() {
            @Override
            public String apply(Long aLong, Long aLong2) throws Exception {
                return aLong + "+" + aLong2 + "=" + (aLong + aLong2);
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

    /**
     * combineLatest操作符演示
     * 把两个Observable产生的结果进行合并，合并的结果组成一个新的Observable。最后一个Observable中的每一个数据项，
     * 都与前面Observable中的最后一项进行相应运算，将生成的结果组成一个新的Observable对象.。
     */
    public static void combineLatestOperator() {
        //被订阅者每隔1000毫秒后开始发射数据，并只取前面3个数据 发射顺序为0 2 4
        Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(3);


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);

        Observable.combineLatest(observable1, observable2, new BiFunction<Long, Long, String>() {
            @Override
            public String apply(Long aLong, Long aLong2) throws Exception {
                return aLong + "+" + aLong2 + "=" + (aLong + aLong2);
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG, "onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG, "onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG, "onError:" + e.toString());
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG, "onComplete");
            }
        });
    }

    /**
     * 在另一个Observable发射的数据定义的时间窗口内，这个Observable发射了一条数据，就结合两个Observable发射的数据。
     */
    public static void joinOperator() {
        Observable<Integer> o1 = Observable.just(1, 2, 3);
        Observable<Integer> o2 = Observable.just(4, 5, 6);

        o1.join(o2, new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                Log.d(XqTag.TAG, "apply1:" + integer);
                return Observable.just(integer + "-o1").delay(200, TimeUnit.MILLISECONDS);
            }
        }, new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                Log.d(XqTag.TAG, "apply2:" + integer);
                return Observable.just(integer + "-o2").delay(800, TimeUnit.MILLISECONDS);
            }
        }, new BiFunction<Integer, Integer, String>() {
            @Override
            public String apply(Integer integer, Integer integer2) throws Exception {
                return integer + " - " + integer2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String value) throws Exception {
                Log.d(XqTag.TAG, "onNext:" + value);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(XqTag.TAG, "onError:" + throwable.toString());
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.d(XqTag.TAG, "onComplete");
            }
        }, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.d(XqTag.TAG, "onSubscribe");
            }
        });
    }

    /**
     *
     */
    public static void concatOperator(){

        //抛错误的Observable
        Observable errorObservable = Observable.error(new Exception("this is end!")).delay(3500, TimeUnit.MILLISECONDS);

        //被订阅者每隔1000毫秒后开始发射数据，并只取前面3个数据 发射顺序为0 2 4
        Observable<Long> observable1 = Observable.interval(0, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2;
                        return result;
                    }
                }).take(3)
                //临时制造失败
                .mergeWith(errorObservable)
                //拦截失败，继续发送下一个数据
                .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Long>>() {
                    @Override
                    public ObservableSource<? extends Long> apply(Throwable throwable) throws Exception {
                        return Observable.empty();
                    }
                });


        //被订阅者延时500毫秒后每隔1000毫秒开始发射数据，并只取前面5个数据 发射顺序是 1 3 5 7 9
        Observable<Long> observable2 = Observable.interval(500, 1000, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        long result = aLong * 2 + 1;
                        return result;
                    }
                }).take(5);

        Observable.concat(observable1,observable2)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.d(XqTag.TAG, "onNext:" + aLong);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete");
                    }
                });

    }
}
